#include "cloud_task.h"
#include "paho_mqtt.h"
#include "cloud_msg.h"
#include "emu_common.h"
#include <ulog.h>

#define MQTT_URI                "tcp://192.168.0.81:1883"   //设置服务器地址
#define MQTT_USERNAME           "sss"                        //代理服务器用户名
#define MQTT_PASSWORD           "ddd"                        //代理服务器密码
#define MQTT_SUBTOPIC           "/mqtt/sub"                   //订阅的 Topic
#define MQTT_PUBTOPIC           "/mqtt/pub"                   //推送的 Topic
#define MQTT_WILLMSG            "Goodbye!"                     //设置断开通知消息

/* define MQTT client context */
static MQTTClient client;
static int is_started = 0;
#define MAX_LEN 2048
static rt_timer_t timer_uploadEmu;
static int s_cloud_status = cloud_init_status;
uint8_t emu_wnParam_upload_url[64]; 
uint8_t emu_emeterParam_upload_url[64];    
uint8_t cloud_setEmuTime_url[64];       
uint8_t emu_resp_setEmuTime_url[64];  
uint8_t cloud_readInv_url[64];          
uint8_t emu_resp_readInv_url[64];     
uint8_t cloud_readEmu_url[64];            
uint8_t emu_resp_readEmu_url[64];      
uint8_t cloud_readAll_url[64];      
uint8_t emu_resp_readAll_url[64];      
uint8_t cloud_writeInv_url[64];       
uint8_t emu_resp_writeInv_url[64];    
uint8_t cloud_writeEmu_url[64];      
uint8_t emu_resp_writeEmu_url[64];      
uint8_t cloud_readEnergyFile_url[64];   
uint8_t emu_resp_EnergyFile_url[64];    
uint8_t cloud_readFaultLog_url[64];     
uint8_t emu_resp_FaultLog_url[64];  

uint8_t cloud_upgrade_url[64];      
uint8_t emu_resp_upgrade_url[64];      
uint8_t cloud_netSetting_url[64];   
uint8_t emu_resp_netSetting_url[64];    
uint8_t cloud_updatePwd_url[64];     
uint8_t emu_resp_updatePwd_url[64];


void mqtt_process(void* param);
void uploadEmu_process(void *parameter);
void cloud_task_init(void)
{
    rt_sprintf(emu_wnParam_upload_url,emu_wnParam_upload_topic,EMU_SN);
    rt_sprintf(emu_emeterParam_upload_url,emu_emeterParam_upload_topic,EMU_SN);
    rt_sprintf(cloud_setEmuTime_url,cloud_setEmuTime_topic,EMU_SN);
    rt_sprintf(emu_resp_setEmuTime_url,emu_resp_setEmuTime_topic,EMU_SN);
    rt_sprintf(cloud_readInv_url,cloud_readInv_topic,EMU_SN);
    rt_sprintf(cloud_readEmu_url,cloud_readEmu_topic,EMU_SN);
    rt_sprintf(emu_resp_readEmu_url,emu_resp_readEmu_topic,EMU_SN);
    rt_sprintf(cloud_readAll_url,cloud_readAll_topic,EMU_SN);
    rt_sprintf(emu_resp_readAll_url,emu_resp_readAll_topic,EMU_SN);
    rt_sprintf(cloud_writeInv_url,cloud_writeInv_topic,EMU_SN);
    rt_sprintf(emu_resp_writeInv_url,emu_resp_writeInv_topic,EMU_SN);
    rt_sprintf(cloud_writeEmu_url,cloud_writeEmu_topic,EMU_SN);
    rt_sprintf(emu_resp_writeEmu_url,emu_resp_writeEmu_topic,EMU_SN);
    rt_sprintf(cloud_readEnergyFile_url,cloud_readEnergyFile_topic,EMU_SN);
    rt_sprintf(emu_resp_EnergyFile_url,emu_resp_EnergyFile_topic,EMU_SN);
    rt_sprintf(cloud_readFaultLog_url,cloud_readFaultLog_topic,EMU_SN);
    rt_sprintf(emu_resp_FaultLog_url,emu_resp_FaultLog_topic,EMU_SN);

    rt_sprintf(cloud_upgrade_url,cloud_upgrade_topic,EMU_SN);
    rt_sprintf(emu_resp_upgrade_url,emu_resp_upgrade_topic,EMU_SN);
    rt_sprintf(cloud_netSetting_url,cloud_netSetting_topic,EMU_SN);
    rt_sprintf(emu_resp_netSetting_url,emu_resp_netSetting_topic,EMU_SN);
    rt_sprintf(cloud_updatePwd_url,cloud_updatePwd_topic,EMU_SN);
    rt_sprintf(emu_resp_updatePwd_url,emu_resp_updatePwd_topic,EMU_SN);

    //初始化mqtt
    /* init condata param by using MQTTPacket_connectData_initializer */
     MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;
     static char cid[20] = { 0 };
     int ret = 0;
     if (is_started)
     {
         LOG_E("mqtt client is already connected.");
         return ;
     }
     /* config MQTT context param */
     {
         client.isconnected = 0;
         client.uri = MQTT_URI;

         /* generate the random client ID */
         rt_snprintf(cid, sizeof(cid), "rtthread%d", rt_tick_get());
         /* config connect param */
         rt_memcpy(&client.condata, &condata, sizeof(condata));
         client.condata.clientID.cstring = cid;
         client.condata.keepAliveInterval = 30;
         client.condata.cleansession = 1;
         client.condata.username.cstring = MQTT_USERNAME;
         client.condata.password.cstring = MQTT_PASSWORD;

         /* config MQTT will param. */
         client.condata.willFlag = 1;
         client.condata.will.qos = 1;
         client.condata.will.retained = 0;
         client.condata.will.topicName.cstring = MQTT_PUBTOPIC;
         client.condata.will.message.cstring = MQTT_WILLMSG;

         /* malloc buffer. */
         client.buf_size = client.readbuf_size = 1024;
         client.buf = rt_calloc(1, client.buf_size);
         client.readbuf = rt_calloc(1, client.readbuf_size);
         if (!(client.buf && client.readbuf))
         {
             LOG_E("no memory for MQTT client buffer!");
             return -1;
         }

         /* set event callback function */
         client.connect_callback = mqtt_connect_callback;
         client.online_callback = mqtt_online_callback;
         client.offline_callback = mqtt_offline_callback;

         /* set subscribe table and event callback */
         client.messageHandlers[0].topicFilter = rt_strdup(MQTT_SUBTOPIC);
         client.messageHandlers[0].callback = mqtt_sub_callback;
         client.messageHandlers[0].qos = QOS1;

         /* set default subscribe event callback */
         client.defaultMessageHandler = mqtt_sub_default_callback;
     }
     /* run mqtt client */
     paho_mqtt_start(&client);
     is_started = 1;



   rt_thread_t thread = rt_thread_create("cloud_task",mqtt_process,RT_NULL,4*4096,25,10);
    /* 创建成功则启动线程 */
   if (thread != RT_NULL)
   {
       rt_thread_startup(thread);
   }
   else
   {
       ret = RT_ERROR;
   }
   LOG_D("finish cloud task \n");
   //定时任务启动
   /*
   rt_system_timer_thread_init();
   timer_uploadEmu = rt_timer_create("timer_uploadEmu",uploadEmu_process,RT_NULL,5000,RT_TIMER_FLAG_PERIODIC|RT_TIMER_FLAG_SOFT_TIMER);
   if (timer_uploadEmu != RT_NULL)
       rt_timer_start(timer_uploadEmu);
   else{
       printf("start timer failed\n");
       return;
   }*/
    rt_thread_t thread_time;// = rt_thread_create("timer_uploadEmu",uploadEmu_process,RT_NULL,1024*1024,25,10);
    /* 创建成功则启动线程 */
   if (thread_time != RT_NULL)
   {
       //rt_thread_startup(thread_time);
   }
   else
   {
       ret = RT_ERROR;
   }
      LOG_D("finish timer_uploadEmu task \n");
    return;
}
int sub_all_cloud_topic();
void mqtt_process(void* param)
{
    //订阅所有的消息,状态机模式
   // mqtt_subscribe(client,);
   while(1)
   {
    switch(s_cloud_status)
    {
        case cloud_online_status:
        {
            //char timer_req_url[64];
            //rt_sprintf(timer_req_url,);
            mqtt_subscribe(&client,"topic1");
            mqtt_subscribe(&client,"topic2");
            sub_all_cloud_topic();
            s_cloud_status = cloud_work_status;
            LOG_D("eeeeee\n");
        }
        break;
        case cloud_work_status:
        {
            //LOG_D("eeeeee2\n");
        }break;
    }
    rt_thread_mdelay(30);
   }
    
}
void uploadEmu_process(void *parameter)
{
    //未测试
    while(1)
    {
        if(s_cloud_status == cloud_work_status){
            msg_dealfuns* node = msg_getfun_node("emu_wn_req","");
            int ret = 0;
            if(node == NULL)
            {
                LOG_E("can not find fun node,not support\n");
                rt_thread_mdelay(1);
                return ;
            }
            //这里使用消息队列接受数据，发送
            char outdata[MAX_LEN];
            rt_memset(outdata,0,MAX_LEN);
            node->upload_handle(outdata);
            uint8_t url[64]={0};
            rt_sprintf(url,emu_wnParam_upload_topic,EMU_SN);
            printf("upload emu url:%s\n",url);
            printf("upload emu url:%s\n",outdata);
            printf("upload emu len:%d\n",rt_strlen(outdata));
            mqtt_publish(url,outdata,rt_strlen(outdata),QOS1);
        }
        rt_thread_mdelay(3000);
    }

}
static void mqtt_sub_callback(MQTTClient *c, MessageData *msg_data)
{
    *((char *)msg_data->message->payload + msg_data->message->payloadlen) = '\0';
    LOG_D("mqtt sub callback: %.*s %.*s",
               msg_data->topicName->lenstring.len,
               msg_data->topicName->lenstring.data,
               msg_data->message->payloadlen,
               (char *)msg_data->message->payload);
    on_cloud_msg(msg_data->topicName->lenstring.data,msg_data->message->payload,msg_data->message->payloadlen);

}
static void mqtt_sub_default_callback(MQTTClient *c, MessageData *msg_data)
{
    *((char *)msg_data->message->payload + msg_data->message->payloadlen) = '\0';
    LOG_D("mqtt sub default callback: %.*s %.*s",
               msg_data->topicName->lenstring.len,
               msg_data->topicName->lenstring.data,
               msg_data->message->payloadlen,
               (char *)msg_data->message->payload);
}
static void mqtt_connect_callback(MQTTClient *c)
{
    LOG_D("inter mqtt_connect_callback!");
}

static void mqtt_online_callback(MQTTClient *c)
{
    LOG_D("inter mqtt_online_callback!");
    s_cloud_status = cloud_online_status;
}

static void mqtt_offline_callback(MQTTClient *c)
{
    LOG_D("inter mqtt_offline_callback!");
    s_cloud_status = cloud_offline_status;
}

static void mqtt_new_sub_callback(MQTTClient *client, MessageData *msg_data)
{
    *((char *)msg_data->message->payload + msg_data->message->payloadlen) = '\0';
    LOG_D("mqtt new subscribe callback: %.*s %.*s",
               msg_data->topicName->lenstring.len,
               msg_data->topicName->lenstring.data,
               msg_data->message->payloadlen,
               (char *)msg_data->message->payload);
    on_cloud_msg(msg_data->topicName->lenstring.data,msg_data->message->payload,msg_data->message->payloadlen);
}


int mqtt_subscribe(MQTTClient* client,char* topic)
{

    if (is_started == 0)
    {
        LOG_E("mqtt client is not connected.");
        return -1;
    }

    return paho_mqtt_subscribe(client, QOS1, topic, mqtt_new_sub_callback);
}
int mqtt_publish(const rt_uint8_t *topic, const rt_uint8_t *data, rt_uint32_t len,rt_uint8_t qos)
{
    paho_mqtt_publish(&client, qos, topic, data);
    return 0;
}

int sub_all_cloud_topic()
{
    mqtt_subscribe(&client,cloud_setEmuTime_url);
    mqtt_subscribe(&client,cloud_readInv_url);
    mqtt_subscribe(&client,cloud_readEmu_url);
    mqtt_subscribe(&client,cloud_readAll_url);
    mqtt_subscribe(&client,cloud_writeInv_url);
    mqtt_subscribe(&client,cloud_writeEmu_url);
    mqtt_subscribe(&client,cloud_readEnergyFile_url);
    mqtt_subscribe(&client,cloud_readFaultLog_url);
    mqtt_subscribe(&client,cloud_upgrade_url);
    mqtt_subscribe(&client,cloud_netSetting_url);
    mqtt_subscribe(&client,cloud_updatePwd_url);
}
int on_cloud_msg(const rt_uint8_t *topic, const rt_uint8_t *data, rt_uint32_t len)
{
    LOG_D("cloud:topic:%s\n", topic);
    LOG_D("data:%s\n", data);
    msg_dealfuns* node = msg_getfun_node("",topic);
    int ret = 0;
    if(node == NULL)
    {
        LOG_D("can not find fun node,not support\n");
        return EMU_ERROR;
    }
    if(node->method == METHOD_UPLOAD)
    {

    }
    else if(node->method == METHOD_REQ)
    {
        char outdata[MAX_LEN];
        rt_memset(outdata,0,MAX_LEN);
        ret = node->req_handle(data,outdata);
        char outurl[64]={0};
        rt_sprintf(outurl,node->respurl,EMU_SN);
        mqtt_publish(outurl,outdata,strlen(outdata),QOS1);
    }
    else if(node->method == METHOD_REP)
    {

    }
    else{
        printf(" not support method\n");

    }
    return ret;
}
